package cn.tedu.process.listener;

import cn.tedu.process.param.ChargingProgressParam;
import cn.tedu.process.param.EMQXMetaData;
import com.alibaba.fastjson2.JSON;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutionException;

/**
 * 测试类而已，实际警告和错误会走rabbitmq，而数据直接接入flink分析了
 */
@Component
public class ProcessNotificationKafkaConsumer {

    /**
     * flink 通过时间滑动窗口 检测到温度高于温度⼤于45度，⼩于48度，由⽤⼾来决定是否继续充电则会发送消息到该队列
     * @param record
     * @throws ExecutionException
     * @throws InterruptedException
     */
    @KafkaListener(topics = "warningChargingTopic",groupId = "Process-consumer-group")
    public void listen(ConsumerRecord<?, String> record) throws ExecutionException, InterruptedException {
        String value = record.value();
        // 监听到flink时间窗口温度48，则webScoket推送给客户警告，可选断开
        System.out.println("I`m warningChargingTopic"+value);
    }

    /**
     * flink 通过时间滑动窗口 检测到温度高于48度则会发送消息到该队列
     * @param record
     * @throws ExecutionException
     * @throws InterruptedException
     */
    @KafkaListener(topics = "errorChargingTopic",groupId = "Process-consumer-group")
    public void listenError(ConsumerRecord<?, String> record) throws ExecutionException, InterruptedException {
        String value = record.value();
        // 监听到flink监控温度高于50，则直接断开，并且告诉用户，则webScoket推送给客户 停止充电
        System.out.println("I`m errorChargingTopic"+value);
    }


    /**
     * 该对列会被flink订阅， 目前测试 emqx作为提供者能不能发送消息
     * @param record
     * @throws ExecutionException
     * @throws InterruptedException
     */
    @KafkaListener(topics = "chargingDataTopic",groupId = "Process-consumer-group")
    public void chargingDataConsumer(ConsumerRecord<?, String> record) throws ExecutionException, InterruptedException {
        String value = record.value();
        // 假设emqxData是已经解析的EMQXMetaData对象
        EMQXMetaData emqxData = JSON.parseObject(value, EMQXMetaData.class);
        // 解析payload
        if (emqxData.getPayload() != null) {
            ChargingProgressParam payload = JSON.parseObject(emqxData.getPayload(), ChargingProgressParam.class);
            emqxData.setPayloadObject(payload); // 如果需要可以再次将其设置回对象中，需要更改payload字段类型回ChargingProgressParam
        }
        System.out.println(emqxData);
        // 监听到flink监控温度高于50，则直接断开，并且告诉用户，则webScoket推送给客户 停止充电

    }
}
